package com.sun.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Slf4j
@Component
public class KafkaConsumerListener {

    @KafkaListener(topics = { "test-security" })
    public void listenRecord(ConsumerRecord<?, ?> record) {
        log.info("record = {}", record);
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());

        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            log.info("topic = {}, partition = {}, message = {}", record.topic(), record.partition(), message);
        }
    }

}
